package com.iddactylogram.server.transfer;

import com.iddactylogram.common.domain.MessageObject;
import com.iddactylogram.common.enumerate.CmdEnums;
import com.iddactylogram.server.transfer.handler.CmdHandler;
import com.iddactylogram.common.utils.MessageParseUtils;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class IdDactylogramServer {

    //监听端口号
    private int port;

    private Selector pipeSelector;
    private Selector acceptSelector;
    //业务处理线程池
    private ThreadPoolExecutor pool;
    //Server端的通道（channel）
    private ServerSocketChannel serverChannel;

    //handler映射
    private final CmdHandlerMapping handlerMapping;
    private final AtomicBoolean isRunning = new AtomicBoolean(true);

    {
        try {
            pipeSelector = Selector.open();
            acceptSelector = Selector.open();
            pool = new ThreadPoolExecutor(16, 64, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public int getPort() {
        return port;
    }

    public IdDactylogramServer() {
        handlerMapping = CmdHandlerMapping.CmdHandlerMappingFactory.getInstance();
    }

    public void start(int port) throws IOException {
        this.port = port;
        serverChannel = ServerSocketChannel.open();
        serverChannel.bind(new InetSocketAddress(port));
        serverChannel.configureBlocking(false);
        serverChannel.register(acceptSelector, SelectionKey.OP_ACCEPT);
        handleRequest();

        while (isRunning.get()) {
            int cnt = acceptSelector.select();
            if (cnt <= 0) {
                continue;
            }
            Iterator<SelectionKey> iterator = acceptSelector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey selectionKey = iterator.next();
                iterator.remove();
                if (selectionKey.isAcceptable()) {
                    ServerSocketChannel serverChannel = (ServerSocketChannel) selectionKey.channel();
                    System.out.println("接受链接=====================" + serverChannel.getLocalAddress());

                    SocketChannel clientChannel = serverChannel.accept();
                    clientChannel.configureBlocking(false);
                    clientChannel.register(pipeSelector, SelectionKey.OP_READ);
                    pipeSelector.wakeup();
                }
            }
        }
    }

    public void stop() {
        //服务端关闭
        try {
            serverChannel.close();
            this.isRunning.set(false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void handleRequest() {
        new Thread(() -> {
            while (isRunning.get()) {
                try {
                    int cnt = pipeSelector.selectNow();
                    if (cnt <= 0) {
                        continue;
                    }
                    Iterator<SelectionKey> iterator = pipeSelector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();
                        SocketChannel channel = (SocketChannel) selectionKey.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(128);
                        int readLen = channel.read(buffer);
                        if (readLen < 0) {
                            System.out.println("断开的链接----");
                            channel.close();
                        } else {
                            distributeAndHandler(channel, buffer);
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    private void distributeAndHandler(SocketChannel channel, ByteBuffer buffer) {
        pool.execute(() -> {
            try {
                MessageObject msgObj = MessageParseUtils.parse(buffer);
                CmdHandler handler = handlerMapping.getHandler(CmdEnums.resolve(msgObj.getCmd()));
                byte[] writeBytes = handler.handle(CmdEnums.resolve(msgObj.getCmd()), msgObj.getMessage());

                ByteBuffer writeBuf = MessageParseUtils.createWriteBuf(writeBytes, msgObj.getCmd());
                writeBuf.flip();
                channel.write(writeBuf);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }
}


